package org.example._02_consumer_model;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        //1.创建一个拉取消息的消费者对象
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("rocketmq-consumer");
        //2.设置名字地址(和 nameser 进行绑定)
        defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");

        //3.绑定消息的主题(订阅你要消费那一个topic的消息),  subscribe:订阅
        defaultMQPushConsumer.subscribe(
                "consumer-model", //第一个参数: 主题
                "*"  //第二个参数: 表达式  用于消息过滤  *表示任何信息都可以进行消费
        );
        //设置消费模式
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);//集群模式
        //4.消费者监听处理消息方法(进行监听,监听指定topic中的消息,如果看到有消息就执行自身业务逻辑)
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * 当监听到指定的 topic 中有消息了就会执行 consumeMessage 方法,一直持续进行监听
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("消费线程:" + Thread.currentThread().getName() +
                            ",消息ID:" + msg.getMsgId() +
                            ",消费者2 消息内容:" + new String(msg.getBody()));
                    //msg.getBody()是一个byte数组,所以转换成字符串
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者(消费者需要持续不断地监听,所以不用关闭)
        defaultMQPushConsumer.start();
    }
}